#include "webserver.h"

struct signal_st st[] =
{
	{ SIGINT, signalHandle },
	{ 0,  NULL }
};

/*
 *@信号处理
 *
 */
void signalHandle(int signo)
{
	int index = 0;
	int threadNum = -1;
	int iRet = RET_ERROR;
	struct systemmsg_st ctlmsg;
	memset(&ctlmsg, 0, sizeof(struct systemmsg_st));
	ctlmsg.msgType = SYSTEMEXIT;
	ctlmsg.msgData = -1;
	switch(signo)
	{
	case  SIGINT:
		//构造消息;
		threadNum = ctx->TotalthreadNum;
		DEBUG_TRACE(("[debug]thread id:[%ld] catch sigint signal!!!!\n", syscall(SYS_gettid)));
		for(index = 0; index < threadNum; index++)
		{
			iRet = sendSystemMsg(ctx, &ctlmsg);

		}
		ctx->bStopFlag = true;
		break;
	default:
		fprintf(stderr, "[error]unknown signal no:%d  in file:[%s] func:[%s] line:[%d]\n", signo, __FILE__,__func__,__LINE__);
		return;
	}
}



/*
 *@brief:create worker thread
 *
 */
int createWorkerThread(struct webServerCtx *serverctx)
{
	int iRet = RET_ERROR;
	if(!serverctx)
	{
		fprintf(stderr,"[error]get unavaliable args in [func]:%s [file]:%s [line]:%d\n",__func__,__FILE__,__LINE__);
		return RET_ERROR;
	}
	int workerThreadNum = serverctx->opt->threadNum;
	int i = 0;
	serverctx->worker=(struct workerPthread_ctx *)malloc(sizeof(struct workerPthread_ctx) * workerThreadNum);
	if(!serverctx->worker)
	{
		fprintf(stderr, "[error]%s in file:[%s] func:[%s] line:[%d]\n", strerror(errno), __FILE__,__func__,__LINE__);
		return RET_ERROR;
	}
	memset(serverctx->worker, 0, sizeof(struct workerPthread_ctx)*workerThreadNum);
	totalthreadNum = serverctx->TotalthreadNum = workerThreadNum;
	serverctx->workqueue = (int *)malloc(sizeof(int)* serverctx->TotalthreadNum);
	for(i = 0; i < workerThreadNum; i++)
	{
		int sock[2] = {0};
		iRet = socketpair(AF_LOCAL, SOCK_STREAM, 0, sock);
		if(iRet == RET_ERROR)
		{
			fprintf(stderr,"[error]%s in  file:[%s] func:[%s] line:[%d]\n", strerror(errno), __FILE__,__func__,__LINE__);
			return RET_ERROR;
		}
		DEBUG_TRACE(("[debug]workqueue:[%d]:[%d]\n", sock[0], sock[1]));
		setfdNoBlock(sock[0]);
		setfdNoBlock(sock[1]);
		serverctx->io->ModEvent(serverctx->io, sock[0], EPOLL_CTL_ADD, EPOLLIN);
		serverctx->worker[i].iotype = serverctx->iotype;
		serverctx->workqueue[i] = sock[0];
		serverctx->worker[i].listensock = sock[1]; //添加到工作线程中;
		serverctx->worker[i].ctx = (void *)serverctx;
        pthread_attr_t attr;
        pthread_attr_init(&attr);
        pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

        iRet = pthread_create(&(serverctx->worker[i].tid), &attr, workerThreadRun, (void *)&(serverctx->worker[i]));
		if(iRet == RET_ERROR)
		{
			fprintf(stderr, "[error]%s  in file:[%s] func:[%s] line:[%d]\n", strerror(errno), __FILE__,__func__, __LINE__);
			return  RET_ERROR;
		}
	}
	return  RET_OK;
}


/*@brief:工作线程回调函数
 *@para:
 *
 */
void * workerThreadRun(void *arg)
{
	int isListenFlag = 0;
#ifdef __DEBUG_ON__
	DEBUG_TRACE(("[debug]worker thread:[%ld] is running!\n",syscall(SYS_gettid)));
#endif
	prctl(PR_SET_NAME, "worker_thread");
	int iRet = RET_ERROR;
	struct workerPthread_ctx * workerctx = (struct workerPthread_ctx *)arg;
	workerctx->tid = syscall(SYS_gettid);
	struct webServerCtx *serverctx = (struct webServerCtx *)(workerctx->ctx);
	size_t maxConnSize = serverctx->opt->maxconnSize;
	iRet = initConnPool(&workerctx, maxConnSize);
	if(iRet == RET_ERROR)
	{
		LogInfo(serverctx, "%s%ld%s\n", "worker thread:[", syscall(SYS_gettid), "] init ConnPool failed!");
		fprintf(stderr, "[error]init ConnPool failed! in file:[%s] func:[%s] line:[%d]\n", __FILE__,__func__,__LINE__);
		return NULL;
	}

	iRet = registerSignalHandle(isListenFlag);
	if(iRet == RET_ERROR)
	{
		LogInfo(serverctx, "%s%ld%s\n", "worker thread:[", syscall(SYS_gettid), "] register signal handle function failed!");
		fprintf(stderr, "[error]register Signal Handle failed! in file:[%s] func:[%s] line:[%d]\n", __FILE__,__func__,__LINE__);
		return NULL;
	}

	iRet = registerIoEvent(&(workerctx->workerIo), workerctx->iotype);
	if(iRet == RET_ERROR)
	{
		LogInfo(serverctx, "%s%ld%s\n", "worker thread:[", syscall(SYS_gettid), "] register Io handle Module failed!");
		fprintf(stderr, "[error]register Io Event failed! in file:[%s] func:[%s] line:[%d]\n", __FILE__,__func__,__LINE__);
		return NULL;
	}

	iRet = httpServerInit(&(workerctx->httpctx), serverctx->opt);
	if(iRet == RET_ERROR)
	{
		LogInfo(serverctx, "%s%ld%s\n", "worker thread:[", syscall(SYS_gettid), "] init Http Server Content failed!");
		fprintf(stderr, "[error]http Server context init failed in file:[%s] func:[%s] line:[%d]\n", __FILE__,__func__,__LINE__);
		return NULL;
	}
	iRet = workerctx->workerIo->ioInit(&(workerctx->workerIo));
	iRet = workerctx->workerIo->ModEvent(workerctx->workerIo, workerctx->listensock, EPOLL_CTL_ADD , EPOLLIN);
	int evCount = 0;
	workerctx->currentConnNum = 0; //当前连接数置0;
	LogInfo(serverctx, "%s%ld%s\n", "worker thread:[", syscall(SYS_gettid), "] is running............!");
	while(!workerctx->bStopFlag)
	{
		evCount = workerctx->workerIo->ioWaitEvent(workerctx->workerIo, DEFAULT_TIMEOUT);
		if(evCount > 0)
		{
			iRet = workerThreadHandleEvent(workerctx, evCount);
			if(iRet == RET_ERROR)
			{
				DEBUG_TRACE(("[debug] occur abnormal httpevent handle!\n"));
			}
		}
		else if(evCount == RET_OK)
		{
			continue;
		}
	}
	workerctx->workerIo->deInit(workerctx->workerIo);
	LogInfo(serverctx, "%s%ld%s\n", "worker thread:[", syscall(SYS_gettid), "] start to exit..........!");
#ifdef __DEBUG_ON__
	DEBUG_TRACE(("[debug]workthread:[%ld] start to exit!!!\n", syscall(SYS_gettid)));
#endif
	totalthreadNum--;
	return workerctx;
}

/*
 *@brief:工作线程的处理函数
 */
int workerThreadHandleEvent(struct workerPthread_ctx *workctx, int evcount)
{
	int iRet = RET_ERROR;
	if(!workctx)
	{
		fprintf(stderr, "[error] get unavaliable args in file[%s] func[%s] line[%d]\n",__FILE__,__func__,__LINE__);
		return iRet;
	}
	int i = 0;
	int iotype = workctx->iotype;
	for ( i = 0; i < evcount; i++)
	{
		switch(iotype)
		{
		case EPOLL_MODULE:
			{
				if( workctx->workerIo->eventset[i].data.fd == workctx->listensock ){
					iRet = getListenThreadMsg(workctx);
				}else{
					iRet = handleHttpEvent(i, workctx);
				}
			}
			break;
		default:
			break;
		}
	}
	return iRet;
}

/*
 *@brief:
 *@para:
 */
int readHttpRequest(struct workerPthread_ctx *workctx, int fd)
{
	int iRet = RET_ERROR;
	if(fd < 0 || !workctx)
	{
		fprintf(stderr, "[error]get unavaliable args in [func]:%s [file]:%s [line]:%d\n", __func__,__FILE__,__LINE__);
		return iRet;
	}
	size_t len = 0;
	unsigned char * buf = workctx->connpool[fd].recvbuf;
	size_t recvSize = workctx->connpool[fd].bufSize;
	len = workctx->connpool[fd].readData(fd, buf, recvSize);
	if(len == 0)
	{
		return RET_OK;
	}
	return len;
}

/*
 *@brief:
 *@para:
 *
 */
int handleHttpEvent(int index, struct workerPthread_ctx *workctx)
{
	int iRet = RET_ERROR;
	if(!workctx)
	{
		fprintf(stderr, "[error]get unavaliable args in [func]:%s [file]:%s [line]:%d\n", __func__,__FILE__,__LINE__);
		return iRet;
	}
	int sended = 0;
	bool isRequest = true;
	struct httpMessage msg;
    memset(&msg, 0, sizeof(struct httpMessage));
	int fd = workctx->workerIo->eventset[index].data.fd;
	struct connection *conn = &(workctx->connpool[fd]);
	for(;;)
	{
		switch(conn->state)
		{
		case CONNECTION_INITIAL:
			conn->state=RECV_REQUEST;
			break;
		case RECV_REQUEST:
			/*
			 *@读取请求;
			 */
			iRet = readHttpRequest(workctx, fd);
			if(iRet > 0)
			{
				DEBUG_TRACE(("[debug]worker Thread:[%ld] get request:%sfd:%d requestLen:%d\n", syscall(SYS_gettid), conn->recvbuf, fd, iRet));
				conn->state = PARSE_REQUEST;
			}else if(iRet == 0){
#ifdef __DEBUG_ON__
				DEBUG_TRACE(("[debug]worker Thread:[%ld] close http request on connection fd:%d\n", syscall(SYS_gettid), fd));
#endif
				conn->state = CONNECTION_CLOSE;
			}
			break;
		case PARSE_REQUEST:
			/*
			 *@解析请求;
			 */
			iRet = httpParseRequest(conn, &msg, isRequest);
            if(iRet > 0){
                conn->state = RECV_REQUEST;
            }else if(iRet == 0){
                conn->state = HANDLE_REQUEST;
            }
			break;
		case HANDLE_REQUEST:
			/*
			 *@处理请求;
			 */
			handleHttpRequest(conn, &msg, workctx->httpctx);
			conn->state = SEND_RESPONSE;
			workctx->workerIo->ModEvent(workctx->workerIo, fd, EPOLL_CTL_MOD, EPOLLOUT|EPOLLET);
			return RET_OK;
		case SEND_RESPONSE:
			/*
			 *@发送响应;
			 */
			sended = conn->writeData(conn->connfd, conn->sendbuf, conn->currDataLen);
			if(sended < 0 )
			{
				fprintf(stderr, "[error]%s in [file]:%s [func]:%s [line]:%d\n", strerror(errno), __FILE__, __func__, __LINE__);
			}
#ifdef __DEBUG_ON__
			DEBUG_TRACE(("[debug]worker Thread:[%ld] send response on fd:%d send data len:%d totalLen:%ld\n", syscall(SYS_gettid), fd, sended, conn->currDataLen));
#endif
			conn->state=CONNECTION_CLOSE;
			break;
		case CONNECTION_CLOSE:
			/*
			 *@关闭网络连接; 目前暂不支持短连接;
			 */
			iRet = delConnection(workctx, fd);
			if( iRet == RET_ERROR )
			{
				fprintf(stderr, "[error]close connection:[%d] failure!\n", fd);
				return iRet;
			}
			return RET_OK;
		default:
			return RET_ERROR;
		}
	}
	return RET_OK;
}


/*
 *@brief:webserver init
 *@para:
 *
 */
int webServerInit(struct webServerCtx **serverctx, int iotype)
{
	int iRet = RET_ERROR;
	int isListenFlag = 1;
	iRet = checkIsNotRootUser();
	if(iRet == RET_ERROR)
	{
		return RET_ERROR;
	}
	struct webServerCtx *tmp = *serverctx;
	tmp = (struct webServerCtx *)malloc(sizeof(struct webServerCtx));
	if(!tmp)
	{
		fprintf(stderr, "[error]%s in file:[%s] func:[%s] line:[%d]\n", strerror(errno),__FILE__, __func__, __LINE__);
		return iRet;
	}
	memset(tmp, 0, sizeof(struct webServerCtx));
	tmp->iotype = iotype;
	tmp->currentIndex = 0;
	/*
	 *@注册信号;
	 */
	iRet = registerSignalHandle(isListenFlag);
	if( iRet == RET_ERROR )
	{
		fprintf(stderr, "[error]%s in file:[%s] func:[%s] line:[%d]\n", strerror(errno), __FILE__,__func__,__LINE__);
		return iRet;
	}
	/*
	 *@初始化配置;
	 */
	iRet = initServerConfigOpt(tmp, CONFIGFILEPATH);
	if(iRet == RET_ERROR)
	{
		fprintf(stderr,"[error]Init Server Config Occur Error! in file:[%s] func:[%s] line:[%d]!\n", __FILE__,__func__,__LINE__);
		goto failed;
	}

	iRet = serverLogInit(tmp);
	if(iRet == RET_ERROR)
	{
		fprintf(stderr,"[error]Init Server Log Module Error! in file:[%s] func:[%s] line:[%d]!\n", __FILE__,__func__,__LINE__);
		goto failed;
	}
	/*
	 *@io注册;
	 */
	iRet = registerIoEvent(&(tmp->io), iotype);
	if(iRet == RET_ERROR)
	{
		fprintf(stderr,"[error]Init Server Config Occur Error! in file:[%s] func:[%s] line:[%d]!\n", __FILE__,__func__,__LINE__);
		goto failed;
	}
	/*
	 *@brief:初始化IO
	 */
	iRet = tmp->io->ioInit(&(tmp->io));
	if(iRet == RET_ERROR)
	{
		fprintf(stderr, "[error]Init IoModel Failed! in file:[%s] func:[%s] line:[%d]!\n", __FILE__,__func__,__LINE__);
		goto failed;
	}
	/*
	 *@创造工作线程
	 */
	iRet = createWorkerThread(tmp);
	if(iRet == RET_ERROR)
	{
		fprintf(stderr,"[error]Init Server Config Occur Error! in file:[%s] func:[%s] line:[%d]!\n", __FILE__,__func__,__LINE__);
		goto failed;
	}
	/*
	 *@brief:
	 */
	iRet = createListenSocket(tmp);
	if(iRet == RET_ERROR)
	{
		fprintf(stderr,"[error]create Listen socket Occur Error! in file:[%s] func:[%s] line:[%d]!\n", __FILE__,__func__,__LINE__);
		goto failed;
	}
	*serverctx = tmp;
	return  RET_OK;
failed:
	*serverctx = tmp;
	webServerDeinit(tmp);
	return  RET_ERROR;
}

/*
 *@brief:WebServer 资源释放
 */
void webServerDeinit(struct webServerCtx *serverctx)
{
	if(!serverctx)
	{
		fprintf(stderr,"[error]get unavaliable args in file:[%s] func:[%s] line:[%d]!\n", __FILE__,__func__,__LINE__);
		return;
	}
    if(serverctx->opt)
	{
		ServerConfigDeinit(serverctx->opt);
	}
	int index = 0;
	for(; index < serverctx->TotalthreadNum; index++)
	{
		ConnPoolDeinit(&(serverctx->worker[index]));
        httpServerDeinit(serverctx->worker[index].httpctx);
	}
    if(serverctx->worker)
	{
        free(serverctx->worker);
        serverctx->worker = NULL;
    }
	if(serverctx->workqueue)
	{
		for(index = 0; index < serverctx->TotalthreadNum; index++)
		{
			if(serverctx->workqueue[index] > 0)
			{
				close(serverctx->workqueue[index]);
			}
		}
		free(serverctx->workqueue);
		serverctx->workqueue = NULL;
	}
	if(serverctx->listenfd)
	{
		close(serverctx->listenfd);
	}
	if(serverctx->log)
    {
		serverLogDeinit(serverctx->log);
	}
    if(serverctx->io)
    {
	    serverctx->io->deInit(serverctx->io);
    }
	if(serverctx)
	{
		free(serverctx);
		serverctx = NULL;
	}
#ifdef __DEBUG_ON__
	DEBUG_TRACE(("[debug]server exit .....!\n"));
#endif
	return ;
}


/*
 *@brief:运行函数;
 */
int webServerRun(struct webServerCtx *serverctx)
{
#ifdef __DEBUG_ON__
	DEBUG_TRACE(("[debug]webserver main thread:[%ld] is running!\n", syscall(SYS_gettid)));
#endif
	LogInfo(serverctx, "%s%ld%s\n", "webserver Listen thread:[", syscall(SYS_gettid),"] is running!");
	int iRet = RET_ERROR;
	int evcount = 0;
	/*
	 *@add fd into set;
	 */
	serverctx->io->ModEvent(serverctx->io, serverctx->listenfd, EPOLL_CTL_ADD, EPOLLIN);
	while(!serverctx->bStopFlag)  //一直在这边循环
	{
		evcount = serverctx->io->ioWaitEvent(serverctx->io, DEFAULT_TIMEOUT);
		if(evcount > 0)
		{
			iRet = mainThreadHandleEvent(serverctx, evcount);
			if(iRet == RET_ERROR)
			{
				continue;
			}
		}
	}
#ifdef __DEBUG_ON__
	DEBUG_TRACE(("[debug]main thread:[%ld] start to exit!!\n", syscall(SYS_gettid)));
#endif
	LogInfo(serverctx, "%s%ld%s\n", "webserver Listen thread:[", syscall(SYS_gettid),"] will exit!");
	while(totalthreadNum>0)
	{
		DEBUG_TRACE(("[debug]TotalthreadNum:%d\n", serverctx->TotalthreadNum));
		sleep(1);
	}
	return RET_OK;
}


/*
 *@brief:
 */
int createListenSocket(struct webServerCtx *serverctx)
{
	int iRet = RET_ERROR;
	if(!serverctx)
	{
		fprintf(stderr, "[error]get unavaliable args:%p in file:[%s] func:[%s] line:[%d]!\n", serverctx, __FILE__,__func__,__LINE__);
		return RET_ERROR;
	}
	serverctx->listenfd = socket(AF_INET, SOCK_STREAM, 0);
	if ( serverctx->listenfd < 0)
	{
		fprintf(stderr, "[error]%s in file:[%s] func:[%s] line:[%d]!\n", strerror(errno), __FILE__,__func__,__LINE__);
		return RET_ERROR;
	}
	LogInfo(serverctx, "%s%d%s%d\n", "listen thread:[", syscall(SYS_gettid), "]create listen socket:", serverctx->listenfd);
	setfdNoBlock(serverctx->listenfd);
	int opt = 1;
	socklen_t optlen = sizeof(opt);
	iRet = setsockopt(serverctx->listenfd, SOL_SOCKET, SO_REUSEADDR, &opt, optlen);
	if(iRet == RET_ERROR)
	{
		fprintf(stderr, "[error]%s in file:[%s] func:[%s] line:[%d]!\n", strerror(errno), __FILE__,__func__,__LINE__);
		return RET_ERROR;
	}
	LogInfo(serverctx, "%s%d%s%d%s\n", "listen thread:[", syscall(SYS_gettid), "] set listen socket:", serverctx->listenfd, "reuse port!");
	memset(&(serverctx->serverAddr), 0, sizeof(struct sockaddr_in));
	serverctx->serverAddr.sin_family = AF_INET;
	serverctx->serverAddr.sin_port = htons(serverctx->opt->listenPort);
	serverctx->serverAddr.sin_addr.s_addr = inet_addr(serverctx->opt->serverIp);
	iRet = bind(serverctx->listenfd, (struct sockaddr *)&(serverctx->serverAddr), sizeof(struct sockaddr_in));
	if(iRet == RET_ERROR)
	{
		fprintf(stderr, "[error]%s in file:[%s] func:[%s] line:[%d]!\n", strerror(errno), __FILE__,__func__,__LINE__);
		return RET_ERROR;
	}
	LogInfo(serverctx, "%s%d%s%s%s%d\n", "listen thread:[", syscall(SYS_gettid), "] bind ip:", serverctx->opt->serverIp, "port:",serverctx->opt->listenPort);
	iRet = listen(serverctx->listenfd, BACKLOG_MAX);
	if(iRet == RET_ERROR)
	{
		fprintf(stderr, "[error]%s in file:[%s] func:[%s] line:[%d]!\n", strerror(errno), __FILE__,__func__,__LINE__);
		return RET_ERROR;
	}
	LogInfo(serverctx, "%s%d%s\n", "listen thread:[", syscall(SYS_gettid), "] start to listen.......");
	DEBUG_TRACE(("[debug]start to listen on port:[%d]  ip:[%s] listenfd:%d\n", serverctx->opt->listenPort, serverctx->opt->serverIp, serverctx->listenfd));
	return RET_OK;
}

/*
 *@brief:
 *@para:
 */
int mainThreadHandleEvent(struct webServerCtx *serverctx, int evCount)
{
	int iRet = RET_ERROR;
	if(!serverctx)
	{
		fprintf(stderr, "[error]get unavaliable args in [file]:%s [func]:%s [line]:%d\n", __FILE__,__func__,__LINE__);
		return iRet;
	}
	int i = 0;
	int iotype = serverctx->iotype;
	for ( i = 0; i < evCount; i++)
	{
		switch(iotype)
		{
		case EPOLL_MODULE:
			{
				if(serverctx->listenfd == serverctx->io->eventset[i].data.fd)
				{
					iRet = getANewConnection(&serverctx);
					if( iRet == RET_ERROR )
					{
						fprintf(stderr, "[error]get A new Connection error in file:[%s] func:[%s] line:[%d]\n",__FILE__,__func__,__LINE__);
						return iRet;
					}
				}
			}
			break;
		default:
			fprintf(stderr, "[error] unknown iotype:%d was found in file:[%s] func:[%s] line:[%d]\n", iotype, __FILE__,__func__,__LINE__);
			return RET_ERROR;
		}
	}
	return RET_OK;
}

/*@brief:
 *@para:
 */
int getANewConnection(struct webServerCtx **server)
{
	int iRet = RET_ERROR;
	struct systemmsg_st msg;
	socklen_t  clilen = sizeof(msg.peeraddr);
	int newconnfd = -1;
	struct webServerCtx *tmp = *server;
	newconnfd = accept(tmp->listenfd, (struct sockaddr *)&(msg.peeraddr), &clilen);  //设置成非阻塞也是有必要的;
	if(newconnfd < 0)
	{
		if(errno == EMFILE || errno == ENFILE)
		{
			LogInfo(tmp, "%s%ld%s\n", "webserver Listen thread[", syscall(SYS_gettid),"] has reach max Connection number!");
			return RET_OK;
		}
		fprintf(stderr, "[error]%s in file:[%s] func:[%s] line:[%d]\n!", strerror(errno), __FILE__,__func__,__LINE__);
		return iRet;
	}
	msg.msgType = NEWCONNECTION;
	msg.msgData = newconnfd;
	iRet = sendSystemMsg(tmp, &msg);
	return iRet;
}


/*
 *@brief:
 */
int getListenThreadMsg(struct workerPthread_ctx *workctx)
{
	size_t len = 0;
	int iRet = RET_ERROR;
	int newconnfd = -1;
#define PEER_CLIENT_IP_ADDRESS 50
	char peerIpAddr[PEER_CLIENT_IP_ADDRESS] = { '\0' };
	struct systemmsg_st  msg = { -1, -1 };
	for(;;)
	{
		len = recv(workctx->listensock, &msg, sizeof(struct systemmsg_st), MSG_DONTWAIT);
		if (len < 0 )
		{
			if(errno == EAGAIN|| errno == EINTR)
			{
				continue;
			}
			return RET_ERROR;
		}
		else if(len == 0)
		{
			return RET_OK;
		}
		else
		{
			switch(msg.msgType)
			{
			case NEWCONNECTION:
				inet_ntop(AF_INET, (struct in_addr *)&(msg.peeraddr.sin_addr),  peerIpAddr, PEER_CLIENT_IP_ADDRESS);
				newconnfd = msg.msgData;
				LogInfo(ctx,"%s%d%s%s\n", "worker thread:[", syscall(SYS_gettid), "] recv system msg : new client connection from Ip:", peerIpAddr);
				iRet = addConnection(workctx, newconnfd);
				break;
			case SYSTEMEXIT:
				LogInfo(ctx,"%s%d%s\n", "worker thread:[", syscall(SYS_gettid), "] recv system msg : server exit!");
				workctx->bStopFlag = true;
				break;
			default:
				LogInfo(ctx,"%s%d%s\n", "worker thread:[", syscall(SYS_gettid), "] recv unknown msg!");
				DEBUG_TRACE(("[error]unknown msg type:%d in file:[%s] func:[%s] line:[%d]\n", msg.msgType, __FILE__,__func__,__LINE__));
				return RET_ERROR;
			}
			return RET_OK;
		}
	}
}

/*
 *@发送函数
 *
 */
int sendSystemMsg(struct webServerCtx *serverctx, struct systemmsg_st *msg)
{
	int iRet = RET_ERROR;
	if(!serverctx)
	{
		fprintf(stderr, "[error]get unavaliable arg in file:[%s] func:[%s] line:[%d]\n",__FILE__,__func__,__LINE__);
		return iRet;
	}
	serverctx->currentIndex = serverctx->currentIndex %(serverctx->TotalthreadNum);
	int sock = serverctx->workqueue[serverctx->currentIndex++];
	for(;;)
	{
		iRet = send(sock, msg, sizeof(struct systemmsg_st), MSG_DONTWAIT);
		if(iRet < 0)
		{
			if(errno == EAGAIN|| errno == EINTR || EWOULDBLOCK)
				continue;
			fprintf(stderr, "[error]send msg error:%s  in file:[%s]  func:[%s] line:[%d]\n", strerror(errno),__FILE__,__func__,__LINE__);
			return RET_ERROR;
		}
		else if(iRet == 0)
		{
			//连接关闭了
			return RET_OK;
		}
		else
		{
			break;
		}
	}
	return iRet;
}

/*
 *@brief:
 *@para:
 */
int registerSignalHandle(int isListenFlag)
{
	int iRet = RET_ERROR;
	struct sigaction  newact, oldact;
	if(isListenFlag)
	{
		newact.sa_handler = signalHandle;
		sigemptyset(&newact.sa_mask);
		newact.sa_flags = SA_SIGINFO;
		int index = 0;
		for(; st[index].signalHandle != NULL; index++)
		{
			iRet = sigaction(st[index].signal, &newact, &oldact);
			if ( iRet == RET_ERROR)
			{
				fprintf(stderr, "[error]%s in file:[%s] func:[%s] line:[%d]\n", strerror(errno), __FILE__, __func__,__LINE__);
				return iRet;
			}
		}
	}
	else
	{
		//工作线程的信号设置;
		struct sigaction sa;
		sa.sa_handler = SIG_IGN;
		sa.sa_flags = 0;
		if(sigemptyset(&sa.sa_mask) == RET_ERROR || sigaction(SIGPIPE, &sa, 0) == RET_ERROR)
		{
			fprintf(stderr, "[error]%s in file:[%s] func:[%s] line:[%d]\n", strerror(errno), __FILE__, __func__,__LINE__);
			return RET_ERROR;
		}
		sigset_t signal_mask;
		sigemptyset(&signal_mask);
		sigaddset(&signal_mask, SIGPIPE);
		iRet = pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
		if(iRet != RET_OK)
		{
			fprintf(stderr, "[error]%s in file:[%s] func:[%s] line:[%d]\n", strerror(errno), __FILE__, __func__,__LINE__);
			return iRet;
		}
	}
	return RET_OK;
}
